package com.cetc.sdp.kmga.cs.util;

import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.function.BiConsumer;

/**
 * spark streaming 任务配置类
 * @description:
 * @author： DengQiang
 * @date: 2017/12/15 10:14
 */
public class StreamingConfiguration {

    private Map<String, String> appProperties = new HashMap<>();
    private Map<String, String> kafkaProperties = new HashMap<>();
    private Map<String, String> hbaseProperties = new HashMap<>();
    private Map<String, String> rabbitmqProperties = new HashMap<>();

    public StreamingConfiguration(String config) {
        parse(config);
    }

    private void parse(String config) {
        DocumentBuilderFactory docBuilderFactory
                = DocumentBuilderFactory.newInstance();
        docBuilderFactory.setIgnoringComments(true);
        InputStream is = StreamingConfiguration.class.getClassLoader().getResourceAsStream(config);
        if (is == null) {
            throw new NullPointerException("配置文件不存在: " + config);
        }
        try {
            DocumentBuilder domBuiler = docBuilderFactory.newDocumentBuilder();
            Document doc = domBuiler.parse(is);

            BiConsumer<Node, Map<String, String>> parseNode = (parent, map) -> {
              for (Node prop = parent.getFirstChild(); prop != null; prop = prop.getNextSibling()) {
                  NodeList nodeList = prop.getChildNodes();
                  String propName = null;
                  String propValue = null;
                  for (int i = 0; i < nodeList.getLength(); i++) {
                      Node n = nodeList.item(i);
                      if (n.getNodeType() == Node.ELEMENT_NODE) {
                          if ("name".equals(n.getNodeName())) {
                              propName = n.getFirstChild().getNodeValue();
                          } else if ("value".equals(n.getNodeName())) {
                              propValue = n.getFirstChild().getNodeValue();
                          }
                      }
                  }
                  if (propName != null && propValue != null) {
                      map.put(propName, propValue);
                  }
              }
            };

            Element root = doc.getDocumentElement();
            if (root != null && ROOT_ELEMENT_NAME.equals(root.getNodeName())) {
                NodeList nodeList = root.getChildNodes();
                if (nodeList != null) {
                    for (int i = 0; i < nodeList.getLength(); i++) {
                        Node node = nodeList.item(i);
                        if (node.getNodeType() == Node.ELEMENT_NODE) {
                            switch (node.getNodeName()) {
                                case APP_ELEMENT_NAME:
                                    parseNode.accept(node, appProperties);
                                    break;
                                case KAFKA_ELEMENT_NAME:
                                    parseNode.accept(node, kafkaProperties);
                                    break;
                                case HBASE_ELEMENT_NAME:
                                    parseNode.accept(node, hbaseProperties);
                                    break;
                                case RABBITMQ_ELEMENT_NAME:
                                    parseNode.accept(node, rabbitmqProperties);
                                    break;
                                default:
                                    break;
                            }
                        }
                    }
                } else {
                    throw new RuntimeException("配置文件格式错误!");
                }
            } else {
                throw new RuntimeException("配置文件格式错误!");
            }
        } catch (ParserConfigurationException e) {
            e.printStackTrace();
        } catch (SAXException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public Map<String, String> getAppProperties() {
        return appProperties;
    }

    public Map<String, String> getKafkaProperties() {
        return kafkaProperties;
    }

    public Map<String, String> getHbaseProperties() {
        return hbaseProperties;
    }

    public Map<String, String> getRabbitmqProperties() {
        return rabbitmqProperties;
    }

    private static final String ROOT_ELEMENT_NAME= "application";
    private static final String APP_ELEMENT_NAME= "app-config";
    private static final String KAFKA_ELEMENT_NAME= "kafka-consumer-config";
    private static final String HBASE_ELEMENT_NAME= "hbase-client-config";
    private static final String RABBITMQ_ELEMENT_NAME= "rabbitmq-consumer-config";
}
